import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson2.TypeReference;
import lombok.Data;
import redis.clients.jedis.Jedis;

import java.lang.reflect.Type;
import java.util.Set;
import java.util.UUID;


/**
 * 问题1：如果队列空了？ 客户端会陷入POP的死循环
 * @param <T>
 */

public class RedisDelayingQueue<T> {

    @Data
    static class TaskItem<T> {
        public String id;
        public T msg;
    }

    private Type TaskType = new TypeReference<TaskItem<T>>() {
    }.getType();

    private Jedis jedis;

    private String queueKey;

    public RedisDelayingQueue(Jedis jedis, String queueKey) {
        this.jedis = jedis;
        this.queueKey = queueKey;
    }


    public void delay(T msg) {
        TaskItem taskItem = new TaskItem();
        taskItem.id = UUID.randomUUID().toString();
        taskItem.msg = msg;
        String s = JSON.toJSONString(taskItem);
        //塞入延时队列，5s后再试
        jedis.zadd(queueKey, System.currentTimeMillis() + 5000, s);
    }


    public void loop() {
        while (!Thread.interrupted()) {
            //只取一条
            Set values = jedis.zrangeByScore(queueKey, 0, System.currentTimeMillis(), 0, 1);
            if (values.isEmpty()){
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    break;
                }
                continue;
            }

           String s = (String) values.iterator().next();
            if (jedis.zrem(queueKey,s)>0){
                TaskItem task = JSON.parseObject(s, TaskType);
                this.handleMsg((String) task.msg);
            }
        }
    }


    public void handleMsg(String msg) {
        System.out.println(msg);
    }


    public static void main(String[] args) {
        Jedis conn = new Jedis("192.168.179.21", 6379);
        conn.select(15);

        RedisDelayingQueue queue = new RedisDelayingQueue<>(conn, "q-demo");

        //生产者
        Thread producer = new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                queue.delay("codehole" + i);
            }
        });

        //消费者
        Thread consumer = new Thread(() -> {
            queue.loop();
        });

        producer.start();
        consumer.start();
        try {
            producer.join();
            Thread.sleep(6000);
            consumer.interrupt();
            consumer.join();
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}
